package com.game.commons.network;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.game.commons.network.packet.ReceivablePacket;

/**
 * 数据包处理器<br>
 * 以正确的顺序处理数据包，每个客户端同一时间只能执行一个数据包。根据接收顺序执行。
 * @author 杜祥
 * @date 2013-6-5
 */
public class PacketProcessor<T extends MMOConnection>
{
	
	private static Logger log = Logger.getLogger(PacketProcessor.class);
	
	/**
	 * 等待执行的包小于这个数量将被kill掉
	 */
	private final static int reduceThreshold = 3;
	
	/**
	 * 等待执行的数据包大于这个数量就会新创建一个包处理任务
	 */
	private final static int increaseThreshold = 50;
	
	
	/**
	 * 同步锁
	 */
	private final Lock lock = new ReentrantLock();
	
	/**
	 * 非空条件
	 */
	private final Condition notEmpty = lock.newCondition();
	
	
	/**
	 * 接收到的包队列
	 */
	private final List<ReceivablePacket<T>> packets = new LinkedList<ReceivablePacket<T>>();
	
	
	/**
	 * 工作的线程列表
	 */
	private final List<Thread> threads = new ArrayList<Thread>();
	
	/**
	 * 最小数量的线程
	 */
	private final int minThreads;
	
	/**
	 * 最大数量的线程
	 */
	private final int maxThreads;
	
	
	/**
	 * 构造一个新的数据包处理器，负责包数据执行。
	 * @param minThreads
	 * @param maxThreads
	 */
	public PacketProcessor(int minThreads, int maxThreads)
	{
		if(minThreads <= 0)
			minThreads = 1;
		if(maxThreads < minThreads)
			maxThreads = minThreads;
		
		this.minThreads = minThreads;
		this.maxThreads = maxThreads;
		
		if(minThreads != maxThreads)
			startCheckerThread();
		
		for(int i = 0; i < this.minThreads; i++)
			newThread();
	}
	
	
	/**
	 * 启动检查线程，负责维护包执行线程列表。
	 */
	private void startCheckerThread()
	{
		new Thread(new CheckerTask(), "PacketProcessor:Checker").start();
	}
	
	
	/**
	 * 创建并启动包处理线程，但只有在线程数小于最大线程数时才会创建新的线程。
	 * @return
	 */
	private boolean newThread()
	{
		if(threads.size() >= maxThreads)
			return Boolean.FALSE;
		
		String name = "PacketProcessor:" + threads.size();
		log.debug("创建新的包处理任务[" + name + "].");
		
		Thread thread = new Thread(new PacketProcessorTask(), name);
		thread.start();
		
		threads.add(thread);
		
		return Boolean.TRUE;
	}
	
	
	/**
	 * 如果当前线程数量大于最小线程数量，调用此方法移除一条线程。
	 */
	private void killThread()
	{
		if(threads.size() > minThreads)
		{
			Thread t = threads.remove(threads.size() - 1);
			
			log.debug("kill PacketProcessor Thread:" + t.getName());
			
			t.interrupt();
		}
	}
	
	/**
	 * 增加一个包到包的执行队列中
	 * @param packet	需要执行的包
	 */
	public final void executePacket(ReceivablePacket<T> packet)
	{
		lock.lock();
		try 
		{
			packets.add(packet);
			notEmpty.signal();
		} 
		finally
		{
			lock.unlock();
		}
	}
	
	
	/**
	 * 返回第一个可用的数据包
	 * @return
	 * 				数据包
	 */
	private ReceivablePacket<T> getFirstAviable()
	{
		for(;;)
		{
			while(packets.isEmpty())
				notEmpty.awaitUninterruptibly();
			
			ListIterator<ReceivablePacket<T>> it = packets.listIterator();
			while(it.hasNext())
			{
				ReceivablePacket<T> packet = it.next();
				
				if(packet.getConnection().tryLoackConnection())
				{
					it.remove();
					return packet;
				}
			}
			notEmpty.awaitUninterruptibly();
		}
	}
	
	
	/**
	 * 包执行任务<br>
	 * 这个任务负责按顺序的执行数据包，保证同一时间只执行一个数据包。
	 * @author 杜祥
	 * @create 2013-6-5
	 */
	private final class PacketProcessorTask implements Runnable
	{

		@Override
		public void run()
		{
			ReceivablePacket<T> packet = null;
			
			for(;;)
			{
				lock.lock();
				try
				{
					if(packet != null)
						packet.getConnection().unLockConnection();
					
					if(Thread.interrupted())
						return;
					
					packet = getFirstAviable();
				}
				finally
				{
					lock.unlock();
				}
				packet.run();
			}
		}
	}
	
	
	/**
	 * 包处理线程队列维护任务<br>
	 * 这个任务会根据包处理的繁忙与空闲来增加或减少线包处理任务数。
	 * @author 杜祥
	 * @create 2013-6-5
	 */
	private final class CheckerTask implements Runnable
	{
		
		/**
		 * 任务检测的间隔时间
		 */
		private final int sleepTime = 60 * 1000;
		
		/**
		 * 最后一次检测的数据包数量
		 */
		private int lastSize = 0;

		@Override
		public void run()
		{
			for(;;)
			{
				try
				{
					Thread.sleep(sleepTime);
				}
				catch (Exception e)
				{
					
				}
				/*等待执行的数据包数*/
				int sizeNow = packets.size();
				
				if(sizeNow < lastSize)
				{
					if(sizeNow < reduceThreshold)
					{
						killThread();
					}
				}
				else if(sizeNow > lastSize && sizeNow > increaseThreshold)
				{
					if(!newThread() && sizeNow > increaseThreshold * 3)
					{
						log.warn("检测到有[" + sizeNow + "]个包在等待执行。可以考虑增大最大线程数或则提升硬件解决问题。");
					}
				}
				lastSize = sizeNow;
			}
		}
	}
	
}
